package flink.demo;

import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;


public class StreamCoGroup_Join_Demo {
    public static void main(String[] args) throws Exception {

        Configuration configuration=new Configuration();

        configuration.setInteger("rest.port", 8822);
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        DataStreamSource<String> leftStream = environment.socketTextStream("localhost", 9999);

        DataStreamSource<String> rightStream = environment.socketTextStream("localhost", 9998);

        SingleOutputStreamOperator<Tuple2<String, String>> mapLeft= leftStream.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String s) throws Exception {
                String[] split = s.split(",");
                return Tuple2.of(split[0], split[1]);
            }
        });


        SingleOutputStreamOperator<Tuple3<String, String,String>> mapRight= rightStream.map(new MapFunction<String, Tuple3<String, String,String>>() {
            @Override
            public Tuple3<String, String,String> map(String s) throws Exception {
                String[] split = s.split(",");
                return Tuple3.of(split[0],split[1],split[2] );
            }
        });

        mapLeft.coGroup(mapRight).where(new KeySelector<Tuple2<String, String>, String>() {
            @Override
            public String getKey(Tuple2<String, String> stringStringTuple2) throws Exception {
                return stringStringTuple2.f0;
            }
        }).equalTo(new KeySelector<Tuple3<String,String,String>, String>() {

            @Override
            public String getKey(Tuple3<String, String, String> stringStringStringTuple3) throws Exception {
                return stringStringStringTuple3.f0;
            }
        }).window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
        .apply(new CoGroupFunction<Tuple2<String, String>, Tuple3<String, String, String>, String>() {

            @Override
            public void coGroup(Iterable<Tuple2<String, String>> iterable, Iterable<Tuple3<String, String, String>> iterable2, Collector<String> collector) throws Exception {

                // 在这里实现  left out  join
                for (Tuple2<String, String> t1 : iterable) {
                    boolean flag = false;
                    for (Tuple3<String, String, String> t2 : iterable2) {
                        // 拼接两表字段输出
                        collector.collect(t1.f0 + "," + t1.f1 + "," + t2.f0 + "," + t2.f1 + "," + t2.f2);
                        flag = true;
                    }

                    if (!flag) {
                        // 如果能走到这里面，说明右表没有数据，则直接输出左表数据
                        collector.collect(t1.f0 + "," + t1.f1 + "," + null + "," + null + "," + null);
                    }
                }

                // TODO  实现以下 right out join


                // TODO 实现  full out join


                // TODO  实现  inner join
            }
        });
        /**
         * 流的 join 算子
         * 案例背景：
         *    流1数据：  id,name
         *    流2数据：  id,age,city
         *    利用join算子，来实现两个流的数据按id关联
         */
        DataStream<String> joinedStream = mapLeft.join(mapRight)
                .where(tp2 -> tp2.f0)
                .equalTo(tp3 -> tp3.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(20)))
                .apply(new JoinFunction<Tuple2<String, String>, Tuple3<String, String, String>, String>() {
                    @Override
                    public String join(Tuple2<String, String> t1, Tuple3<String, String, String> t2) throws Exception {
                        return t1.f0 + "," + t1.f1 + "," + t2.f0 + "," + t2.f1 + "," + t2.f2;
                    }
                });

        environment.execute();
    }
}
